package com.atguigu.cn.dataStream.api

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.functions.{AssignerWithPeriodicWatermarks, AssignerWithPunctuatedWatermarks, KeyedProcessFunction}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.watermark.Watermark
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector

/**
  * @author: yangShen
  * @Description: 窗口的操作
  * @Date: 2020/4/16 22:38 
  */
object WindowTest {
  def main(args: Array[String]): Unit = {
    //1.environment
    val environment = StreamExecutionEnvironment.getExecutionEnvironment
    environment.setParallelism(1)
    //environment.setStreamTimeCharacteristic(时间语义类型)：水位线默认配置如下(可点进去查看源码)
      //ProcessingTime ： watermark = 0
      //EventTime ： watermark = 200L
    environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)   //设置时间语义类型
    //改变上面的默认配置
    //environment.getConfig.setAutoWatermarkInterval(100L)

    // 2.source
//    val inputStream = environment.readTextFile("D:\\my\\myGit\\mayun\\miaohui8023\\my-flink\\flink-tutorial\\src\\main\\resources\\sensor.txt")
    val inputStream = environment.socketTextStream("10.16.26.16",7777)

    // 3.transform 操作，但没有标记返回类型
    val dataStream = inputStream.map(data => {
      val dataArray = data.split(",")
      SensorReading(dataArray(0).trim, dataArray(1).trim.toLong, dataArray(2).trim.toDouble)
    })
      //设置时间戳
      //方式一(简单实现)：升序数据，不设置延时 到点就发车，从字段中抽取
//      .assignAscendingTimestamps(_.timestamp * 1000)
      //方式二：周期性时间戳(自定义)
//      .assignTimestampsAndWatermarks(new MyAssignTimestampAndWatermarks())
      //方式三：可断点时间戳(时间戳)
//      .assignTimestampsAndWatermarks(new MyAssignWithPunctuatedWatermark())
      //方式四：处理乱序数据，延迟1秒钟上涨水位的操作(当时间到第11秒时，则认为第10秒为处理时间)
        //EventTime 的时钟设置 = (element.timestamp * 1000) - 1
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.seconds(1)) {
      override def extractTimestamp(element: SensorReading): Long = element.timestamp * 1000
    })



    //转化流程：DataStream-分区->KeyedStream-时间窗->WindowedStream-聚合(要根据窗口区分开)->DataStream
    //统计10秒内最小的温度：默认时间语义为Processing Time,处理数据的时间远小于10秒，时间还没到数据就处理完了，会出现问题 比如没有数据输出，所以需要：socketTextStream
     val minTemperature = dataStream.map(t => (t.id, t.temperature))
       //元组，不能用字段要用下划线
       .keyBy(_._1)
        //到了设定的时间再去该window中获取数据，如果是一个有限流且未到设定的时间，则获取不到数据，
          //比如上面：environment.readTextFile("D:\\my\\myGit\\mayun\\miaohui8023\\my-flink\\flink-tutorial\\src\\main\\resources\\sensor.txt")
       .timeWindow( Time.seconds(10))   //开时间窗口(滚动窗口)，默认的时间语义是Processing Time：定义去哪个窗口
       //统计15秒的最小温度，隔五秒种输出一次
//       .timeWindow(Time.seconds(15),Time.seconds(5))//滑动窗口

       //设置时间为北京时间：东八区-8
       //.window(SlidingEventTimeWindows.of(Time.seconds(15),Time.seconds(5), Time.hours(-8)))
       .reduce( (data1, data2) => (data1._1, data1._2.min(data2._2) ))    //定义窗口关闭做什么操作：用reduce做增量聚合

    minTemperature.print(" min temperature test")
//    dataStream.print("input data")


    //底层API：ProcessFunction
    dataStream.keyBy(_.id).process(new MyProcess())


    environment.execute("window test")

  }
}

//时间戳自定义：周期性生成时间戳
class MyAssignTimestampAndWatermarks() extends AssignerWithPeriodicWatermarks[SensorReading]{
  //60秒
  val bound  = 6000
  var maxTimestamp: Long = Long.MaxValue

  override def getCurrentWatermark: Watermark = new Watermark(maxTimestamp - bound)

  override def extractTimestamp(t: SensorReading, l: Long): Long = {
    maxTimestamp = maxTimestamp.max(t.timestamp * 1000)
    t.timestamp * 1000
  }
}
//时间戳自定义：可打断的 时间戳
class MyAssignWithPunctuatedWatermark extends AssignerWithPunctuatedWatermarks[SensorReading]{
  override def checkAndGetNextWatermark(lastElement: SensorReading, extractedTimestamp: Long): Watermark = new Watermark(extractedTimestamp)

  override def extractTimestamp(element: SensorReading, previousElementTimestamp: Long): Long = element.timestamp * 1000
}
class MyProcess extends KeyedProcessFunction[String, SensorReading, String]{
  //流中的每一个元素都会调用这个方法
  override def processElement(value: SensorReading, ctx: KeyedProcessFunction[String, SensorReading, String]#Context, out: Collector[String]): Unit = {
    //获取当前元素的key
    val key = ctx.getCurrentKey
    //获取当前元素的时间戳
    val long = ctx.timestamp()
    //注册一个2000毫秒的事件时间定时器
    ctx.timerService().registerEventTimeTimer(2000)
    //获取当前的EventTime(水位线)
    //ctx.timerService().currentWatermark()
    //删除时间事件定时器
    //ctx.timerService().deleteEventTimeTimer(2000)

    //ctx.timerService().registerProcessingTimeTimer(2000)
    //ctx.timerService().currentProcessingTime()
    //ctx.timerService().deleteProcessingTimeTimer(2000)

  }
}

